package com.aiwiown.snackmq.broker;

import com.aiwiown.snackmq.broker.config.BrokerConfig;
import com.aiwiown.snackmq.broker.handler.MessageDispatcher;
import com.aiwiown.snackmq.broker.handler.MessageHandlerFactory;
import com.aiwiown.snackmq.broker.transaction.TransactionCheckerService;
import com.aiwiown.snackmq.cluster.config.ClusterConfig;
import com.aiwiown.snackmq.cluster.metadata.MetadataManager;
import com.aiwiown.snackmq.cluster.raft.SnackMQRaftNodeConfig;
import com.aiwiown.snackmq.cluster.raft.SnackMQStateMachine;
import com.aiwiown.snackmq.cluster.raft.manager.SnackMQRaftNodeManager;
import com.aiwiown.snackmq.cluster.timer.TimerScannerService;
import com.aiwiown.snackmq.common.cluster.BrokerId;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.common.protocol.TopicPartition;
import com.aiwiown.snackmq.management.BrokerMetrics;
import com.aiwiown.snackmq.management.scaling.AutoScalerService;
import com.aiwiown.snackmq.management.scaling.ScalingPolicy;
import com.aiwiown.snackmq.network.config.NetworkConfig;
import com.aiwiown.snackmq.network.connection.ConnectionManager;
import com.aiwiown.snackmq.network.server.NettyServer;
import com.aiwiown.snackmq.storage.partition.LogPartition;
import com.aiwiown.snackmq.storage.service.PartitionedStorageService;
import com.aiwiown.snackmq.storage.service.ZeroCopyPartitionedStorageService;
import com.aiwiown.snackmq.storage.timer.TimerMessageStore;
import com.aiwiown.snackmq.storage.transaction.TransactionLog;
import com.alipay.sofa.jraft.rpc.RaftRpcServerFactory;
import com.alipay.sofa.jraft.rpc.RpcServer;
import com.alipay.sofa.jraft.util.Endpoint;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * 负责创建和初始化 Broker 所有核心组件的工厂类。
 * 将组件的创建逻辑与 BrokerServer 的协调逻辑分离。
 */
@Slf4j
public class BrokerComponentFactory {

    private final BrokerConfig brokerConfig;
    private final Map<TopicPartition, LogPartition> activePartitions;
    private final BrokerId selfId;

    public BrokerComponentFactory(BrokerConfig brokerConfig, Map<TopicPartition, LogPartition> activePartitions, BrokerId selfId) {
        this.brokerConfig = brokerConfig;
        this.activePartitions = activePartitions;
        this.selfId = selfId;
    }

    public RpcServer createRaftRpcServer() {
        return RaftRpcServerFactory.createRaftRpcServer(new Endpoint(brokerConfig.getHost(), brokerConfig.getRaftPort()));
    }

    public MetadataManager createMetadataManager(RpcServer raftRpcServer) {
        ClusterConfig clusterConfig = createClusterConfig();
        SnackMQRaftNodeConfig metaRaftConfig = createRaftConfig(MetadataManager.METADATA_GROUP_ID, "meta");
        return new MetadataManager(metaRaftConfig, raftRpcServer, clusterConfig);
    }

    public PartitionedStorageService createStorageService() {
        // 【修复】调用 PartitionedStorageService 的新构造函数，传入 brokerConfig 中定义的存储路径
        return new PartitionedStorageService(brokerConfig.getStorePath(), activePartitions);
    }

    public ZeroCopyPartitionedStorageService createZeroCopyStorageService() {
        if (brokerConfig.isZeroCopyEnabled()) {
            log.info("Zero-copy storage service enabled");
            return new ZeroCopyPartitionedStorageService(activePartitions);
        }
        log.info("Zero-copy storage service disabled");
        return null;
    }

    public TimerMessageStore createTimerMessageStore() {
        try {
            return new TimerMessageStore(
                    brokerConfig.getStorePath(),
                    BrokerServer.wheelSizeInSeconds,
                    brokerConfig.getTimerLogSegmentSize()
            );
        } catch (IOException e) {
            throw new RuntimeException("Failed to initialize TimerMessageStore", e);
        }
    }

    public TransactionLog createTransactionLog() {
        try {
            return new TransactionLog(brokerConfig.getStorePath());
        } catch (IOException e) {
            throw new RuntimeException("Failed to initialize TransactionLog", e);
        }
    }

    public SnackMQStateMachine createSnackMQStateMachine(PartitionedStorageService storageService, MetadataManager metadataManager, TimerMessageStore timerMessageStore) {
        String currentBrokerAddress = selfId.getHost() + ":" + selfId.getPort();
        return new SnackMQStateMachine(storageService, metadataManager, currentBrokerAddress, brokerConfig.getRaftDataPath(), activePartitions, timerMessageStore);
    }

    public SnackMQRaftNodeManager createDataRaftNodeManager(SnackMQStateMachine stateMachine, RpcServer raftRpcServer) {
        ClusterConfig clusterConfig = createClusterConfig();
        SnackMQRaftNodeConfig dataRaftConfig = createRaftConfig(brokerConfig.getRaftGroupId(), "data");
        return new SnackMQRaftNodeManager(dataRaftConfig, stateMachine, raftRpcServer, clusterConfig);
    }

    public TimerScannerService createTimerScannerService(TimerMessageStore timerMessageStore, PartitionedStorageService storageService, SnackMQRaftNodeManager dataRaftNodeManager) {
        return new TimerScannerService(
                timerMessageStore,
                storageService,
                dataRaftNodeManager,
                BrokerServer.wheelSizeInSeconds
        );
    }

    public TransactionCheckerService createTransactionCheckerService(TransactionLog transactionLog) {
        return new TransactionCheckerService(
                transactionLog,
                ConnectionManager.getInstance(),
                brokerConfig.getTransactionTimeoutMs(),
                brokerConfig.getTransactionCheckIntervalMs()
        );
    }

    public AutoScalerService createAutoScalerService(BrokerMetrics brokerMetrics) {
        ScalingPolicy scalingPolicy = ScalingPolicy.builder()
                .enabled(brokerConfig.isAutoScalingEnabled())
                .highTpsThreshold(brokerConfig.getScalingHighTps())
                .lowTpsThreshold(brokerConfig.getScalingLowTps())
                .checkIntervalSeconds(60)
                .build();
        return new AutoScalerService(scalingPolicy, brokerMetrics);
    }

    public NettyServer createNettyServer(MetadataManager metadataManager, SnackMQRaftNodeManager dataRaftNodeManager, PartitionedStorageService storageService, ZeroCopyPartitionedStorageService zeroCopyStorageService, SnackMQStateMachine stateMachine, BrokerMetrics brokerMetrics, AutoScalerService autoScalerService, TransactionLog transactionLog) {
        List<MessageHandler> handlers = MessageHandlerFactory.createHandlers(
                metadataManager,
                dataRaftNodeManager,
                storageService,
                zeroCopyStorageService,
                stateMachine,
                brokerMetrics,
                autoScalerService,
                transactionLog,
                brokerConfig,
                selfId
        );
        MessageDispatcher dispatcher = new MessageDispatcher(handlers);
        NetworkConfig networkConfig = new NetworkConfig();
        networkConfig.setPort(brokerConfig.getServerPort());
        networkConfig.setBossThreads(brokerConfig.getBossThreads());
        networkConfig.setWorkerThreads(brokerConfig.getWorkerThreads());
        return new NettyServer(networkConfig, dispatcher);
    }

    private ClusterConfig createClusterConfig() {
        ClusterConfig clusterConfig = new ClusterConfig();
        clusterConfig.setHost(brokerConfig.getHost());
        clusterConfig.setRaftPort(brokerConfig.getRaftPort());
        clusterConfig.setBusinessPort(brokerConfig.getServerPort());
        clusterConfig.setRaftInitialConf(brokerConfig.getRaftInitialConf());
        clusterConfig.setRaftElectionTimeoutMs(brokerConfig.getRaftElectionTimeoutMs());
        clusterConfig.setRaftAddressToBusinessPortMap(brokerConfig.getRaftAddressToBusinessPortMap());
        return clusterConfig;
    }

    private SnackMQRaftNodeConfig createRaftConfig(String groupId, String subDir) {
        SnackMQRaftNodeConfig raftConfig = new SnackMQRaftNodeConfig();
        raftConfig.setGroupId(groupId);
        String baseRaftPath = brokerConfig.getRaftDataPath();
        String specificRaftPath = baseRaftPath + File.separator + subDir;
        raftConfig.setRaftDataPath(specificRaftPath);
        raftConfig.setHost(brokerConfig.getHost());
        raftConfig.setRaftPort(brokerConfig.getRaftPort());
        String initialConfStr = brokerConfig.getRaftInitialConf() == null || brokerConfig.getRaftInitialConf().isEmpty()
                ? brokerConfig.getHost() + ":" + brokerConfig.getRaftPort()
                : brokerConfig.getRaftInitialConf();
        raftConfig.setInitialConf(initialConfStr);
        raftConfig.setElectionTimeoutMs(brokerConfig.getRaftElectionTimeoutMs());
        raftConfig.setSnapshotIntervalSecs(brokerConfig.getSnapshotIntervalSecs());
        return raftConfig;
    }
}